package com.mjf.day2

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

/**
 * 富函数
 *  富函数是函数类的增强版，可以有生命周期方法，还可以获取运行时上下文，在运行时上下文可以对state进行操作
 *  flink有状态的流式计算，做状态编程，就是基于RichFunction的
 */
object RichMapFunctionExample {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream: DataStream[SensorReading] = env.addSource(new SensorSource)

    // 使用自定义类方式重写map方法
    stream.map(new MyRichMapFunction).print()

    env.execute("RichMapFunctionExample")

  }

  class MyRichMapFunction extends RichMapFunction[SensorReading, String] {

    // 做一些初始化操作，比如连接数据库。只执行一次
    override def open(parameters: Configuration): Unit = {}

    override def map(value: SensorReading): String = value.id

    // getRuntimeContext()方法提供了函数的RuntimeContext的一些信息，例如函数执行的并行度，任务的名字，以及state状态
//    getRuntimeContext.getState()

    // 生命周期中的最后一个调用的方法，做一些清理工作。
    override def close(): Unit = {}

  }
}
